package com.study.sequence.common;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;

import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by 张波 on 2020/5/23.
 */
@Data
@Slf4j
public class SequenceDTO implements Serializable {

    public static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(2, 20, 30, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(10), new MyThreadFactory(), new ThreadPoolExecutor.DiscardPolicy ());
    /*双buffer序列池A*/
    private SequenceBuffer sequenceBuffer_A;
    /*双buffer序列池B*/
    private SequenceBuffer sequenceBuffer_B;
    /*当前使用的buffer*/
    private SequenceBuffer currentBuffer;
    /*策略转换器，为true时使用序列池A，为false时使用序列池B*/
    private AtomicBoolean bufferFlag;
    /*刷新时间*/
    private LocalDateTime flushTime = LocalDateTime.now();
    /*刷新比例值*/
    @Value("${SEQUENCE.PROPORTION:0.2f}")
    private float proportion;
    @Value("SEQUECNT.FLUSH.TIME.SECOND:30")
    private int flushTimeSecond;
    /*是否已经偿试刷新序列池*/
    private AtomicBoolean threadFlus;
    /*是否已经对序列池进行了刷新*/
    private AtomicBoolean isFlus;
    /*刷新规则服务类*/
    private FlushService flushService;
    /*当前池标记:true:A池，false:B池*/
    private AtomicBoolean current;
    /*累计器，用于统计当前获取了多少个sequence*/
    private AtomicLong currentCount;
    /*指数器，用于标识在累计器获取了多少个sequence后需要检测自动刷新机制*/
    private long flushCount;

    private ReentrantLock swapLock = new ReentrantLock();

    private ReentrantLock flushLock = new ReentrantLock();
    /*业务key*/
    private String businessKey;

    /**
     * 初始化
     *
     * @param businessKey
     */
    public SequenceDTO(String businessKey, FlushService flushService) {
        sequenceBuffer_A = new SequenceBuffer("A", businessKey);
        sequenceBuffer_B = new SequenceBuffer("B", businessKey);
        this.businessKey = businessKey;
        this.flushService = flushService;
        this.threadFlus = new AtomicBoolean(false);
        this.isFlus = new AtomicBoolean(false);
        this.bufferFlag = new AtomicBoolean(true);
        this.current = new AtomicBoolean(true);
        this.currentCount = new AtomicLong(0);

    }

    /**
     * 校验是否已经初始化
     *
     * @return
     */
    public boolean isInit() {
        if (bufferFlag.get()) {
            currentBuffer = sequenceBuffer_A;
        } else {
            currentBuffer = sequenceBuffer_B;
        }
        return currentBuffer.getFlush().get();
    }

    /**
     * 池对象初始化
     *
     * @param valueSize    序列位长度
     * @param currentValue 序列位值
     * @param stepValue    缓存位步长值
     * @return
     */
    public void init(int valueSize, int currentValue, int stepValue) {
        if (bufferFlag.get()) {
            /*如果bufferFlag为true，则说明本次需要进行初始化的为池A*/
            sequenceBuffer_A.init(valueSize, currentValue, stepValue, true);
            sequenceBuffer_B.init(valueSize, currentValue, stepValue, false);
        } else {
            /*如果bufferFlag为false,则说明本次需要进行初始化的为池B*/
            sequenceBuffer_B.init(valueSize, currentValue, stepValue, true);
            sequenceBuffer_A.init(valueSize, currentValue, stepValue, false);
        }
        this.flushCount = (long) (proportion * stepValue);
    }

    public String getSequence() {
        return getSequence(null);
    }

    /**
     * 获取序列编号
     *
     * @param prefix
     * @return
     */
    public String getSequence(String prefix) {
        String value = currentBuffer.getSequence(prefix);
        if (currentCount.incrementAndGet() >= this.flushCount) {
            if (current.get()) {
                /*如果是true，则从A池中获取，到指数器位置了需要刷新B*/
                checkFlush(sequenceBuffer_B);
            } else {
                /*否则从B池中获取，到指数器位置了需要刷新A*/
                checkFlush(sequenceBuffer_A);
            }
        }
        /*则说明已经需要更换buffer源了*/
        if (value == null) {
            try {
                swapLock.lock();
                value = syncRetryGetSequence(prefix);
            } finally {
                swapLock.unlock();
            }
        }
        return value;
    }

    /**
     * 校验是否需要刷新缓存池对象
     *
     * @param buffer
     */
    private void checkFlush(SequenceBuffer buffer) {
        if (!isFlus.get()) {
            /*校验是否满足刷新条件，需要则刷新B池*/
            if (currentBuffer.isFlus(proportion) && !buffer.getFlush().get()) {
                /*校验是否已经有在进行刷新中的任务*/
                if (threadFlus.compareAndSet(false, true) || (threadFlus.get() && LocalDateTime.now().minusSeconds(flushTime.getSecond()).getSecond() >= flushTimeSecond)) {
                /*如果没有刷新的任务，或者如果flushTimeSecond秒了，还没有更新掉isFlus，则说明刷新失败了*/
                    THREAD_POOL_EXECUTOR.execute(new SequenceFlushRunable(this));
                }
            }
        }
    }

    /**
     * CAS换池
     *
     * @param prefix
     * @return
     */
    private String syncRetryGetSequence(String prefix) {
        String value = currentBuffer.getSequence(prefix);
        if (null == value) {
            /*说明还没换源，需要换源*/
            swapBuffer();
        } else {
            return value;
        }

        for (int i = 0; i < 3; i++) {
            /*偿试重新获取*/
            value = this.getSequence(prefix);
            if (null == value) {
                continue;
            }
            return value;
        }
        if (null == value) {
            log.info("currentBuffer:{},A.flshFlag:{},B.flushFlag:{},current:{}", currentBuffer.getName(), sequenceBuffer_A.getFlush().get(), sequenceBuffer_B.getFlush().get(), this.current.get());
            throw new RuntimeException("本次获取数据序列失败");
        }
        return value;
    }

    /**
     * 换池
     */
    private void swapBuffer() {
        if (this.current.get()) {
            /*当前用的是池A*/
            currentBuffer = sequenceBuffer_B;
            log.info("换池A->B");
            if (!sequenceBuffer_B.getFlush().get()) {
                /*如果池B还没准备好，则开始串行执行任务刷新操作*/
                log.info("串行更新池B");
                this.flushService.flush(this);
            }
            this.current.compareAndSet(true, false);
        } else {
            /*当前用的是池B*/
            currentBuffer = sequenceBuffer_A;
            log.info("换池B->A");
            if (!sequenceBuffer_A.getFlush().get()) {
                /*如果池A还没准备好，则开始串行执行任务刷新操作*/
                log.info("串行更新池A");
                this.flushService.flush(this);
            }
            this.current.compareAndSet(false, true);
        }
        /*换池，则刷新状态需要变更*/
        isFlus.compareAndSet(true, false);
        currentCount = new AtomicLong(0);
    }

    /**
     * 休眠最多5秒，检查是否刷新完成
     *
     * @param sequenceBuffer
     */
//    private void waitTime(SequenceBuffer sequenceBuffer) {
//        int count = 5;
//        while (count-- > 0) {
//            try {
//                TimeUnit.SECONDS.sleep(1);
//            } catch (InterruptedException e) {
//                log.info("休眠异常：{}", e);
//            }
//            if (sequenceBuffer.getFlush().get()) {
//                return;
//            }
//        }
//        throw new RuntimeException("获取序列号错误，未能刷新序列号对象");
//    }
}
